package com.atguigu.gmall.realtime.util

import java.util

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.text.Text
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.index.query.{BoolQueryBuilder, MatchQueryBuilder, RangeQueryBuilder}
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.aggregations.InternalOrder.Aggregation
import org.elasticsearch.search.aggregations.{AggregationBuilders, Aggregations, BucketOrder, InternalOrder}
import org.elasticsearch.search.aggregations.bucket.terms.{ParsedStringTerms, Terms, TermsAggregationBuilder}
import org.elasticsearch.search.aggregations.metrics.{AvgAggregationBuilder, ParsedAvg}
import org.elasticsearch.search.builder.SearchSourceBuilder
import org.elasticsearch.search.fetch.subphase.highlight.{HighlightBuilder, HighlightField}
import org.elasticsearch.search.sort.SortOrder

object MyEsUtil {

  var esClient: RestHighLevelClient = build()


  def build(): RestHighLevelClient = {
    val config = PropertiesUtil.load("config.properties")
    val host = config.getProperty("es.host")
    val port = config.getProperty("es.port")
    val restClientBuilder: RestClientBuilder = RestClient.builder(new HttpHost(host, port.toInt))
    val restHighLevelClient = new RestHighLevelClient(restClientBuilder)
    restHighLevelClient
  }

  //单条数据的写入
  def save(source: AnyRef): Unit = {

    val indexRequest = new IndexRequest() //写入的索引？ 写入的值？
    indexRequest.index("movie_test0701_20211203")

    val movieJson: String = JSON.toJSONString(source, new SerializeConfig(true))
    indexRequest.source(movieJson, XContentType.JSON)
    esClient.index(indexRequest, RequestOptions.DEFAULT) //向索引写入数据
  }

  //批量写入 幂等？ 通过增加id 作为唯一标识  实现幂等
  def bulkSave(list: List[(AnyRef, String)],indexName:String): Unit = {
    if (list != null && list.size > 0) {
      println(s"即将保存 list :${list.size}" )
      val bulkRequest = new BulkRequest() //bulkrequest实际上就是由多个单条indexRequest组合
      for ((source, id) <- list) {
        val indexRequest = new IndexRequest() //写入的索引？ 写入的值？
        indexRequest.index(indexName)
        val movieJson: String = JSON.toJSONString(source, new SerializeConfig(true))
        indexRequest.source(movieJson, XContentType.JSON)

        indexRequest.id(id)

        bulkRequest.add(indexRequest)
      }

      esClient.bulk(bulkRequest, RequestOptions.DEFAULT)
    }
  }

  def main(args: Array[String]): Unit = {
    //    val movie = Movie("109","梅艳芳")
    //    save(movie)
    //    val movies = List((Movie("110","梅兰芳"),"110"),(Movie("111","沙丘"),"111"))
    //     bulkSave(movies)

    //  search()

    searchAgg()
    destory()

  }


  case class Movie(var id: String, var movie_name: String)


  def destory(): Unit = {
    esClient.close()
    esClient = null
  }


  def search(): Unit = {


    val searchRequest = new SearchRequest()
    searchRequest.indices("movie_index0701")

    val searchSourceBuilder = new SearchSourceBuilder()
    //查询条件

    val boolQueryBuilder = new BoolQueryBuilder()

    //过滤
    val rangeQueryBuilder: RangeQueryBuilder = new RangeQueryBuilder("doubanScore").gt(5.0)
    boolQueryBuilder.filter(rangeQueryBuilder)

    //匹配
    val matchQueryBuilder = new MatchQueryBuilder("name", "red sea")
    boolQueryBuilder.must(matchQueryBuilder)

    searchSourceBuilder.query(boolQueryBuilder)

    //高亮
    val highlightBuilder: HighlightBuilder = new HighlightBuilder().field("name")
    searchSourceBuilder.highlighter(highlightBuilder)

    //分页
    searchSourceBuilder.from(0)
    searchSourceBuilder.size(20)
    //排序
    searchSourceBuilder.sort("doubanScore", SortOrder.DESC)

    searchRequest.source(searchSourceBuilder)

    val searchResponse: SearchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT)

    val total: Long = searchResponse.getHits.getTotalHits.value
    println(s"共查询$total 条数据")
    val hits: Array[SearchHit] = searchResponse.getHits.getHits
    for (searchHit <- hits) {
      println(searchHit.getSourceAsString)
      //处理高亮结果
      val fields: util.Map[String, HighlightField] = searchHit.getHighlightFields
      val highlightField: HighlightField = fields.get("name")
      val fragments: Array[Text] = highlightField.getFragments
      val highlightText: Text = fragments(0)
      println(highlightText)

    }
  }

  def getClient(): RestHighLevelClient ={
    esClient
  }


  def searchAgg(): Unit = {

    val searchSourceBuilder = new SearchSourceBuilder()


    // 先 分组  再聚合
    //1 分组
    val aggregationBuilder: TermsAggregationBuilder = AggregationBuilders
      .terms("groupby_actor")
      .field("actorList.name.keyword").size(1000)

    aggregationBuilder.order(BucketOrder.aggregation("avg_score", false))


    //2 聚合
    val avgAggBuilder: AvgAggregationBuilder = AggregationBuilders.avg("avg_score").field("doubanScore")

    //3 把聚合操作放在分组中作为 subAggregation
    aggregationBuilder.subAggregation(avgAggBuilder)

    searchSourceBuilder.aggregation(aggregationBuilder)


    val searchRequest: SearchRequest = new SearchRequest().indices("movie_index0701").source(searchSourceBuilder)

    val searchResponse: SearchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT)

    // 提取聚合结果
    val groupby_actorTerms: ParsedStringTerms = searchResponse.getAggregations.get[ParsedStringTerms]("groupby_actor")
    val buckets: util.List[_ <: Terms.Bucket] = groupby_actorTerms.getBuckets
    import scala.collection.JavaConverters._
    for (termBucket <- buckets.asScala) {
      val key: AnyRef = termBucket.getKey
      val count: Long = termBucket.getDocCount
      val aggregations: Aggregations = termBucket.getAggregations
      val avg_scoreAvg: ParsedAvg = aggregations.get[ParsedAvg]("avg_score")
      val avgscore: Double = avg_scoreAvg.getValue
      println(s"演员$key 共演了$count 影片 ，平均得分:$avgscore ")

    }

  }

}
